package com.gitee.xhs.saveAndModifyingPointWriter.keyedState.aggregatingState;

import com.gitee.xhs.keyedState.aggregatingState.AggregateFunctionDao;
import com.gitee.xhs.keyedState.aggregatingState.AggregatingStateConstant;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.state.api.functions.KeyedStateBootstrapFunction;

/**
 * @author BigHead
 * @date 2022/7/12 10:38
 * @description AggregatingStateNewSavePointsFunction
 */
public class AggregatingStateNewSavePointsFunction extends KeyedStateBootstrapFunction<Long, Transaction> {

    private transient AggregatingState<Transaction, AggregateFunctionDao> aggregatingState;

    @Override
    public void open(Configuration parameters) throws Exception {
        aggregatingState = getRuntimeContext().getAggregatingState(AggregatingStateConstant.AGGREGATING_STATE_DESCRIPTOR);
    }

    @Override
    public void processElement(Transaction value, KeyedStateBootstrapFunction<Long, Transaction>.Context ctx) throws Exception {
        aggregatingState.add(value);
    }
}
